# MariposaDB Node

# Copyright 2008 Matt Mankins <mankins @ gmail . com>
#   
#    Licensed under the Apache License, Version 2.0 (the "License");
#   you may not use this file except in compliance with the License.
#   You may obtain a copy of the License at
#
#  http://www.apache.org/licenses/LICENSE-2.0
#   
#   Unless required by applicable law or agreed to in writing, software
#   distributed under the License is distributed on an "AS IS" BASIS,
#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#   See the License for the specific language governing permissions and
#   limitations under the License.


package MariposaDB::Node;

our @ISA = qw/Object::Event/; # gives us event support

use strict;

use FindBin qw($Bin);
use lib "$Bin/../lib";

use Coro;
use Coro::AnyEvent;

use Log::Log4perl qw(:easy);
use Data::Dumper;

use Digest::SHA1 qw(sha1_hex);
use Math::BigInt;
use AnyEvent::Socket;
use Object::Event;

use MariposaDB::Utils;
use MariposaDB::Data::Messages;

# User Defines That Should be in the Config
my $MAX_IDLE_TIME = 30;
    
sub new
{
    my $class = shift;
    my ($self)  = @_;

    $self = $self || {};
    bless ($self, $class);

    unless ($self->{'node_name'})
    {
	# by default, this is built by physical node + : + virtual node number -- should just be consistent
	$self->{'node_name'} = $self->{'physical_node'} . ":". $self->{'virtual_node_number'};
    }

    unless ($self->{'position'})
    {
	$self->_position_init();
    }

    # error check
    unless ($self->{'config'} && $self->{'position'} && ($self->{'node_name'} ne ":") )
    {
	DEBUG "ERROR STARTING NODE: ".$self->{'node_name'};
	die "error starting node";
    }

    # setup the callbacks
    $self->_setup_callbacks();

    # this is where the connections live
    $self->{'connections'} = {};

    # start up the server
    $self->{'server'} = $self->_init_server();

    return $self;
}

sub _setup_callbacks
{
    my $self = shift;
    
    # Start up Event Processing callbacks
    $self->reg_cb ('command_init' => sub { DEBUG "New command started"; });
    $self->reg_cb 
	(
	 'on_gossip' => sub 
	 {
	     my (undef,$gossip_request,$connection) = @_;

	     async_pool
	     {
		 DEBUG "GR: $gossip_request, con: $connection";

		 # we got a gossip request.

		 my $GossipResponse = MariposaDB::Data::Messages::Gossip->new();
		 
		 # figure out what type it is: full, or digest.
		 if ( ($gossip_request->has_full()) && ($gossip_request->full()) )
		 {
		     # They want the full list.
		     my $roster = $self->{'gossip'}->get_roster();
		     
		     if ($GossipResponse->unpack($roster->{'roster'}))
		     {
			 DEBUG "roster unpacked ok";
		     }
		     else
		     {
			 DEBUG "roster unpack error";
		     }
		 }
		 else
		 {
		     # just a digest
		     my $roster = $self->{'gossip'}->get_roster("roster_digest");
		     
		     $GossipResponse->set_node_list_digest($roster->{'roster_digest'});
		 }
		 
		 my $Response = MariposaDB::Data::Messages->new();
		 $Response->set_gossip($GossipResponse);
		 
		 my $serialized = $Response->pack();
		 my $one_time_delim = &MariposaDB::Utils::one_time_delim();
		 
		 $self->{'connections'}->{$connection}->{'handle'}->push_write("msg $one_time_delim\r\n$serialized\r\nmsg $one_time_delim\r\n");
	     };
	 }
	);

    $self->reg_cb ('message_complete' => sub 
		   {
		       my (undef,$message,$connection) = @_;
		       DEBUG "got '$message' for $connection";
		
		       my $status = "";
		       my $start_time = AnyEvent->time;
		       my $Messages = MariposaDB::Data::Messages->new;
		       if ( $Messages->unpack($message) ) 
		       {
			   $status = "ok";
		       } else 
		       {
			   $status = "NOT OK";
		       }
		       my $timer = (AnyEvent->time - $start_time)*1000;
		       DEBUG "Parse: $status in $timer ms";

		       unless ($status eq "ok")
		       {
			   DEBUG "not doing more message parsing.";
			   return;
		       }

		       DEBUG $Messages->debug_string();
		       DEBUG "**";

		       my $gossip = $Messages->gossip();		       
		       my $has_gossip = $Messages->has_gossip();
		       if ($has_gossip)
		       {

			   DEBUG $gossip;
			   DEBUG $Messages;
			   DEBUG $gossip->has_node_list();


			   $self->event('on_gossip' => $gossip, $connection);
		       }

		       DEBUG "done message parsing";
		   });

}

sub _init_server
{
    my $self = shift;

    # TODO: This is big and ugly. fix me

    my $port = $self->node_port();
    my $host = $self->node_host();
    DEBUG "NODE starting at position:".$self->{'position'}. " on port $port, host $host\n";
    
    # setup the server
    $self->{'server'} = AnyEvent::Socket::tcp_server $host, $port, 
    # on connect
    sub 
    {
	my ($client_sock, $host, $port) = @_;

	# on client connect this gets executed
	DEBUG "Got new client connection: $host:$port\n";

	# get new connection counter
	my $this_connection = $self->{'connection_count'}++;

	# on session start,we set the last active to now
	$self->{'connections'}->{$this_connection}->{'last_activity'} = AnyEvent->time;

	# setup an anyevent handle for this connection
	$self->{'connections'}->{$this_connection}->{'handle'} =
	    AnyEvent::Handle->new 
	    (
	     'fh' => $client_sock,
	     'on_eof' => sub { INFO "client connection $host:$port: eof\n" },
	     'on_error' => sub { WARN "Client connection error: $host:$port: $!\n" },
	    );

	# set maximum idle timeout
	$self->{'maximum_idle_timer'} = AnyEvent->timer 
	    (
	     'after' => 5,
	     'interval' => 5, 
	     'cb' => sub 
	     {
		 unless ( ($self->{'connections'}->{$this_connection}) && ($self->{'connections'}->{$this_connection}->{'handle'}) )
		 {
		     undef $self->{'maximum_idle_timer'};

		     return;
		 }

		 if ( (AnyEvent->time - $self->{'connections'}->{$this_connection}->{'last_activity'}) > $MAX_IDLE_TIME)
		 {
		     $self->{'connections'}->{$this_connection}->{'handle'}->push_write ("err Timeout\015\012");
		     $self->{'connections'}->{$this_connection}->{'handle'}->on_drain 
			 (
			  # and close it up
			  sub 
			  {
			      $self->{'connections'}->{$this_connection}->{'handle'}->fh->close; 
			      undef $self->{'connections'}->{$this_connection}; # all gone, hopefully
			  }
			 );
		 }
	     }
	    );

	DEBUG "starting $this_connection";
	$self->_start_command($this_connection);
    };

    return $self->{'server'};
}

sub _start_command
{
    my $self = shift;

    my ($connection) = @_;

    DEBUG "Starting command for connection: $connection";

    # we only process one command at a time per connection
    # so we are either in the middle of processing a command, or are waiting for a new command (undef)
    $self->{'connections'}->{$connection}->{'server_state'} = undef;  # in this case, we're starting, so we're waiting

    $self->_clear_buffers($connection);

    $self->event('command_init');

    # and start reading/parsing
    DEBUG "conn now: $connection";
    $self->_read_next_chunk($connection);
}

sub _clear_buffers
{
    my $self = shift;

    my ($connection) = @_;

    $self->{'connections'}->{$connection}->{'message'} = undef;
}

sub _read_next_chunk
{
    my $self = shift;
    my ($connection)  = @_;
    
    DEBUG "Connection: $connection\n";

    $self->{'connections'}->{$connection}->{'handle'}->push_read( sub { DEBUG "rnc";  $self->_read_command_chunk(@_, $connection); } );
}

sub _read_command_chunk
{
    my ($self, $handle, $connection) = @_; #undef, $handle?

    DEBUG "RBUF?";
    
    return unless defined $handle->{'rbuf'};

    DEBUG "Got buf: ".$handle->{'rbuf'};
#    open (R, ">>/tmp/r");
#    print R $handle->{'rbuf'};
#    close R;

    $self->{'connections'}->{$connection}->{'last_activity'} = AnyEvent->time;

    foreach my $line (split /\r*\n/, $handle->{'rbuf'})
    {
	# if we're in the middle of reading a command, keep at it
	if ($self->{'connections'}->{$connection}->{'server_state'} eq "msg")
	{
	    if (substr($line,0,4) ne "msg ")
	    {
		# this is a continuation of a previous read
		$self->{'connections'}->{$connection}->{'message'} .= $line."\n";
	    }
	    elsif ($line eq ("msg ".$self->{'connections'}->{$connection}->{'delim'}))
	    {
		# end of message
		chop $self->{'connections'}->{$connection}->{'message'}; # remove last \n

		DEBUG "End of message, sending event.";
		DEBUG "msg:'".$self->{'connections'}->{$connection}->{'message'}."'";

		# send the message to others for processing
		$self->event('message_complete', $self->{'connections'}->{$connection}->{'message'}, $connection);
	
		# reset
		$self->{'connections'}->{$connection}->{'delim'} = undef;
		$self->{'connections'}->{$connection}->{'message'} = "";
		$self->{'connections'}->{$connection}->{'server_state'} = undef;
	    }
	    else
	    {
		# the line just happens to start with "msg "
		# this is a continuation of a previous read                                                                                                                                   
                DEBUG "line just happens to have msg a start";
		$self->{'connections'}->{$connection}->{'message'} .= $line."\n";
	    }
	}
	else
	{
	    # no state, see what the command is
	    if ($line =~ /^msg (.+)\s*(.*)/)
	    {
		my $delim = $1;
		chomp $delim;
		$self->{'connections'}->{$connection}->{'delim'} = $delim; # the current msg delim
		DEBUG "Got Delim: '$delim'";
		$self->{'connections'}->{$connection}->{'server_state'} = "msg"; # waiting for a full message

		$self->{'connections'}->{$connection}->{'message'} = "";

		# TODO: if we have any envelope options such as compression, parse them here.
		
		# TODO: Set alarms to timeout a read that takes too long
	    }
	    elsif ($line =~ /^quit/i)
	    {
		$self->{'connections'}->{$connection}->{'handle'}->push_write ("err bye bye\015\012");
		$self->{'connections'}->{$connection}->{'handle'}->on_drain 
		    (
		     # and close it up
		     sub 
		     {
			 $self->{'connections'}->{$connection}->{'handle'}->fh->close; 
			 undef $self->{'connections'}->{$connection}; 
		     }
		    );
	    }
	    elsif ($line =~ /^help/i)
	    {
		my $help_msg = <<_HELP_
		    
 Commands: msg, help, quit, stats, err
		 
 msg:
 idea: on client, first generate random string of N non-null, non-whitespace bytes. 
 This is the "one time delimeter". A message is enveloped by the command + space + this one time delim at the 
 start and at the end of the message.  In the first envelope line, there can be optional "header vars" in URL encoding format.
 these optional header vars might be used for later expansion of things like compress=gzip&content-length=12345&md5= ...

 example:
 msg QOJZIYQSUUWQUGTXRCWEWPEVYDQNTEPXTRIGCOCSOIVRINESKWZVNCTULSEMAIIIHOPMUXWKERWTFJVSUXPAMEGSXFXEMLOCRXJT
 Hello World
 msg QOJZIYQSUUWQUGTXRCWEWPEVYDQNTEPXTRIGCOCSOIVRINESKWZVNCTULSEMAIIIHOPMUXWKERWTFJVSUXPAMEGSXFXEMLOCRXJT

 The client (sender) must examine each line to determine if the actual one time delim appears in the message,
 and if so MUST immediately abort and drop the connection.  It is presumed that the client will continue to retry
 until the entire message has been transmitted along with the trailing end delim.

 No content length is required with this scheme, although there is a price of a strcmp for each line,
 this method should be tested...to determine optimal length of one time delim

 help:
  some in-line documentation of the protocol and what can be done when debugging
		 
 quit:
  terminate the connection
		 
 stats:
  basic statistics

 err:
  a one line error message, usually used by the server to send an unsolicited response to the client
  ex: err Timeout

_HELP_
;
		
		$self->{'connections'}->{$connection}->{'handle'}->push_write($help_msg);
		$self->{'connections'}->{$connection}->{'server_state'} = undef;
	    }
	    else
	    {
		$self->{'connections'}->{$connection}->{'handle'}->push_write("err unknown command\r\n");
		$self->{'connections'}->{$connection}->{'server_state'} = undef; # resets
	    }
	}
    }

    delete $handle->{'rbuf'};
    $self->_read_next_chunk($connection) if (defined $self->{'connections'}->{$connection}->{'handle'});
}

sub _position_init
{
    my $self = shift;
    
    # setup virtual node position

    my $digest = Digest::SHA1::sha1_hex($self->{'node_name'});
    my $exact_location = Math::BigInt->new("0x$digest"); # this is where this hash is on the circle, but we break up into buckets, so we really want the bucket.
    # should be less than 2^160
    # this is our node_id, the name of the node. Which is not necessarily == the position of the node, as the position is rounded down to the nearest bucket start.
    $self->{'node_id'} = "0x$digest";

    # now we see which bucket it's in
    my $el = $exact_location->copy();

    # how many keys fit in each bucket?
    $self->{'buckets'} = $self->{'config'}->get("buckets") || 11;

    $self->{'bucket_size'} = Math::BigInt->new(2);
    $self->{'bucket_size'} = $self->{'bucket_size'}->bpow( (160-$self->{'buckets'}) ); # 2^160, total circle

    # how far past our bucket start point is this node?
    my $leftover = $exact_location->bmod($self->{'bucket_size'});
    DEBUG "left: ".$leftover."< ".$self->{'bucket_size'}." ?\n";
    DEBUG "exact: ".$el;

    $self->{'position'} = $el->bsub($leftover); # back up

    DEBUG "Virtual node init: ".$self->{'position'}."\n";
    
    return $self->{'position'};
}

sub node_port
{
    # each virtual node listens to its own port
    my $self = shift;
    my ($set_port) = @_;

    if ($set_port)
    {
	$self->{'port'} = $set_port;
    }

    if ($self->{'port'})
    {
	# already have it
	return $self->{'port'};
    }

    # determine by our virtual node number
    my ($iface,$port) = split(':',$self->{'config'}->get("membership"));

    $self->{'port'} = $port - 1 + $self->{'virtual_node_number'}; # virtual node numbers start at 1
    return $self->{'port'};
}

sub node_host
{
    # each virtual node listens to its own host
    my $self = shift;
    my ($set_host) = @_;

    if ($set_host)
    {
	$self->{'host'} = $set_host;
    }

    if ($self->{'host'})
    {
	# already have it
	return $self->{'host'};
    }

    # determine by our virtual node number
    my ($host,$port) = split(':',$self->{'config'}->get("membership"));

    $self->{'host'} = $host;
    return $self->{'host'};
}

1;
